package com.atbeijing.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atbeijing.gmall.realtime.common.GmallConfig;
import com.atbeijing.gmall.realtime.utils.DimUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * 维度流输出到hbase
 */
public class DimSink extends RichSinkFunction<JSONObject> {
    private Connection conn;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        Class.forName(GmallConfig.PHOENIX_DRIVER);
        conn = DriverManager.getConnection(GmallConfig.PHOENIX_SERVER);

    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        //写入hbase哪个表
        String sinkTable = value.getString("sink_table").toUpperCase();
        //获取数据,不为空就写入hbase
        JSONObject data = value.getJSONObject("data");
        if (data!=null && data.size()>0){
            //拼接sql
            String upsertSql = genUpsertSql(sinkTable,data);
            PreparedStatement ps = null;
            try {
                ps = conn.prepareStatement(upsertSql);
                ps.executeUpdate();
                //手动提交事务
                conn.commit();
                System.out.println("向Phoenix中插入数据的SQL：" + upsertSql);
            } catch (SQLException e) {
                e.printStackTrace();
                throw new RuntimeException("向Phoenix中插入数据失败");
            } finally {
                if(ps != null){
                    ps.close();
                }
            }
        }

        //现实中业务表一般不会有删除,删除也是逻辑删除,相当于更新.所以要么增加,要么修改,而增加redis中本来就没有缓存
        if(value.getString("type").equals("update")||value.getString("type").equals("delete")){
            //清空当前数据在Redis中的缓存
            DimUtil.deleteCached(sinkTable,data.getString("id"));
        }
    }

    //拼接插入到Phoenix表的SQL
    private String genUpsertSql(String sinkTable, JSONObject data) {
        String upsertSql = "upsert into "+GmallConfig.HBASE_SCHEMA
                +"."+sinkTable
                +" ("+ StringUtils.join(data.keySet(),",") +") " +   //(a,b,c)
                " values(\'"+StringUtils.join(data.values(),"\',\'")+"\')"; //('a','b','c')
        return upsertSql;
    }
}
